package com.broada.apm.es.util;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.broada.apm.Constants;

/**
 * 工厂类
 * @author jianfei.xu
 * @date   2016年10月13日
 *
 */
public class BulkProcessorFactory {

	public static BulkProcessor makeBulkProcessor(Client client, String moduleName) {
		BulkProcessor bulkProcessor = BulkProcessor.builder(client, new BulkProcessorListener(moduleName))
				.setBulkActions(Constants.thriftBatchSize)
				.setBackoffPolicy(BackoffPolicy.constantBackoff(delayTimeValue, Constants.maxRetryTimesWhenValidateTimeOut))
				.setBulkSize(byteSizeValue)
				.setFlushInterval(flushTimeValue)
				.build();
		
		return bulkProcessor;
	}
	
	private static class BulkProcessorListener implements BulkProcessor.Listener {

		@Override
		public void beforeBulk(long executionId, BulkRequest request) {
			
		}

		@Override
		public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
			if(response.hasFailures()) {
				LOG.error(moduleName, response.buildFailureMessage());
			}
		}

		@Override
		public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
			LOG.error(moduleName, failure);
		}
		
		public BulkProcessorListener(String moduleName) {
			this.moduleName = moduleName;
		}
		
		private static final Logger LOG = LoggerFactory.getLogger(BulkProcessorListener.class);
		
		private String moduleName; //模块名称

	}
	
	private static TimeValue flushTimeValue = new TimeValue(10 * 1000); //10s
	
	private static TimeValue delayTimeValue = new TimeValue(5 * 1000); //5s
	
	private static ByteSizeValue byteSizeValue = new ByteSizeValue(10, ByteSizeUnit.MB);
	
}
